package com.tanghao.flink.demo.distinct;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer010;
import org.apache.flink.streaming.util.serialization.SerializationSchema;
import org.apache.flink.streaming.util.serialization.SimpleStringSchema;

import java.io.InputStream;
import java.util.Properties;

/**
 * Created by Bill on 2019-5-9.
 */
//https://blog.csdn.net/weixin_33781606/article/details/86125054
public class KafkaMessageStreaming {
    public static void main(String[] args) throws Exception {
        String inTopic = args[0];
        String outTopic = args[1];
        String redisUrl = args[2];
        String jobName = args[3];
        String kafkaProps = args[4];

        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(300000);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        Configuration conf = new Configuration();
        conf.setString("redis_url",redisUrl);
        conf.setString("job_name",jobName);
        env.getConfig().setGlobalJobParameters(conf);

        Properties props = new Properties();
        InputStream in = KafkaMessageStreaming.class.getClassLoader().getResourceAsStream(kafkaProps);
        props.load(in);

        FlinkKafkaConsumer010<String> consumer =
                new FlinkKafkaConsumer010<>(inTopic, new SimpleStringSchema(), props);
        consumer.assignTimestampsAndWatermarks(new MessageWaterEmitter());

        FlinkKafkaProducer010<Tuple2<String, String>> producer =
                new FlinkKafkaProducer010<Tuple2<String, String>>(outTopic, new SerializationSchema<Tuple2<String, String>>() {
                    @Override
                    public byte[] serialize(Tuple2<String, String> element) {
                        return (element.f0+","+element.f1).getBytes();
                    }
                }, props);

        env.addSource(consumer)
                .flatMap(new MessageSplitter())
                .keyBy(1, 2, 3)
                .flatMap(new UniqAnalysis())
                .addSink(producer);

        env.execute(jobName);
    }
}